package org.zfes.snowier.xyz.live;

import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.zfes.snowier.common.socket.data.MsgData;
import org.zfes.snowier.common.socket.device.group.GroupAttr;
import org.zfes.snowier.common.socket.device.sender.INettyMsgSender;
import org.zfes.snowier.common.socket.device.status.DeviceStaus;
import org.zfes.snowier.core.util.UUIDUtil;
import org.zfes.snowier.core.util.ZAssert;
import org.zfes.snowier.core.util.ZObjectUtil;
import org.zfes.snowier.core.util.ZStrUtil;
import org.zfes.snowier.xyz.live.dto.LiveStatus;

import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
@Component
public class LiveRequestSimpleManager implements ILiveRequestManager, InitializingBean  {

	@Autowired IVideoStatusManager videoStatusManager;
	
	//<deviceNum#channelId，Set<String>>
	private static final Map<String,Set<String>> channelRequestMap = Maps.newConcurrentMap();
	//<requestId，deviceNum#channelId>
	private static final Map<String,String> requestChannelMap = Maps.newConcurrentMap();
	//<requestId，lastTime>
	private static final Map<String,Long> requestLasTimeMap = Maps.newConcurrentMap();
	
	private static final Long deadTime = 5*1000L;

	private final ReentrantLock lock = new ReentrantLock();
	 
	private static ScheduledExecutorService scheduledExecutor;
		
	private AtomicBoolean detectStarted  = new AtomicBoolean(false);
	
	@Override
	public void afterPropertiesSet() throws Exception {
		startDetect();
	}
	 
	 
	public  synchronized void startDetect() {
		if(detectStarted.get()) {
			return;
		}
		getExecutor().scheduleAtFixedRate(()->{
			Long now =System.currentTimeMillis();
			
			requestLasTimeMap.entrySet().forEach(en->{
				
				if(now - en.getValue() > deadTime) {
					String requestId=en.getKey();
					String device_chanl =requestChannelMap.get(requestId);
					if(ZStrUtil.hasText(device_chanl)) {
						  String deviceNum=device_chanl.split("#")[0];
						  String channelId=device_chanl.split("#")[1];
						onRequestDead(deviceNum,channelId,requestId);
					}
					
				}
			});
			
		},5,2,TimeUnit.SECONDS);
		detectStarted.set(true);
	}
	
	@Override
	public boolean isChannelHaveRequest(String deviceNum, String channelId) {
		if(ZStrUtil.hasNoText(deviceNum)||ZStrUtil.hasNoText(channelId)) {
			return false;
		}
		AtomicBoolean haveRequest=new AtomicBoolean(false);
		String device_chanl = constructKey( deviceNum,  channelId);
		channelRequestMap.compute(device_chanl, (key,val)->{
			haveRequest.set(val!=null&&!val.isEmpty());
			return val;
		});
		return haveRequest.get();
	}
	//支持心跳，第一次产生request
	@Override
	public void onRequestHeatBeat(String deviceNum, String channelId,String requestId){
		if(ZStrUtil.hasNoText(requestId)||ZStrUtil.hasNoText(deviceNum)||ZStrUtil.hasNoText(channelId)) {
			return;
		}
		lock.lock();
		String device_chanl = constructKey( deviceNum,  channelId);
		try {
			channelRequestMap.compute(device_chanl, (key,val)->{
				if(val==null) {
					val=Sets.newHashSet();
				}
				val.add(requestId);
				return val;
			});
			requestChannelMap.put(requestId, constructKey( deviceNum,  channelId));
			requestLasTimeMap.put(requestId, System.currentTimeMillis());
			
		}finally {
			lock.unlock();
		}
	}
	
	@Override
	public void onRequestDead(String deviceNum, String channelId,String requestId){
		if(ZStrUtil.hasNoText(requestId)||ZStrUtil.hasNoText(deviceNum)) {
			return;
		}
		
		lock.lock();
		try {
			String device_chanl = constructKey( deviceNum,  channelId);

			requestChannelMap.remove(requestId);
			requestLasTimeMap.remove(requestId);
			
			Set<String> reqSet = channelRequestMap.get(device_chanl);
			if(reqSet!=null) {
				reqSet.remove(requestId);
			}
			if(reqSet==null||reqSet.isEmpty()) {//如果集合为空，则发送停止推流命令
				videoStatusManager.liveChannelStop(deviceNum, channelId);
			}
			
			
		}finally {
			lock.unlock();
		}
	}
	
	@Override
	public long getReauestLastHeartBeatTime(String requestId) {
		if(ZStrUtil.hasNoText(requestId)) {
			return 0L;
		}
		return requestLasTimeMap.get(requestId);
	}
	
	@Override
	public Set<String> getChannelRequestSet(String deviceNum, String channelId){
		String key=constructKey( deviceNum,  channelId);
		if(ZStrUtil.hasNoText(key)) {
			return Sets.newHashSet();
		}
		 Set<String>  requestSet=channelRequestMap.get(key);
		return requestSet==null?Sets.newHashSet():requestSet;
	}
	
	@Override
	public Optional<String> getRequestChannel(String requestId){
		if(ZStrUtil.hasNoText(requestId)) {
			return Optional.empty();
		}
		String  requestSet= requestChannelMap.get(requestId);
		return Optional.ofNullable(requestSet);
	}
	

	private ScheduledExecutorService getExecutor() {
		if(scheduledExecutor==null) {
			scheduledExecutor = Executors.newScheduledThreadPool(2,new ThreadFactoryBuilder().setNameFormat("-request-live-heartbeats-detect--pool-%d").build());
		}
		return scheduledExecutor;
	}
	

	private String constructKey(String deviceNum, String channelId) {
		
		return deviceNum+"#"+channelId;
	}
	
}
